package net.bwie.dwd;


import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;


public class DwdEntranceLog {
    public static void main(String[] args) throws Exception {
        // 1.表执行环境
        TableEnvironment tableEnv = getTableEnv();

        // 2.输入表 读取Kafka数据
        createInputTable(tableEnv);

        // 3.数据处理
        Table resultTable = handle(tableEnv);

        // 4.映射表 写入Kafka表中
        createOutputTable(tableEnv);

        // 5.保存数据
        saveToSink(tableEnv, resultTable);

    }

    private static void saveToSink(TableEnvironment tableEnv, Table resultTable) {
        // todo 创建临时视图 然后写入Kafka中
        tableEnv.createTemporaryView("EntranceGuardLog", resultTable);
        tableEnv.executeSql(
                "insert into dwd_entrance_guard_log_kafka select * from EntranceGuardLog"
        );
    }

    private static void createOutputTable(TableEnvironment stream) {
        // todo 创建Kafka表
        stream.executeSql(
                "create table dwd_summary_cars_log(\n" +
                        "    id STRING,\n" +
                        "    owerId BIGINT,\n" +
                        "    opTime TIMESTAMP(3),\n" +
                        "    cId BIGINT,\n" +
                        "    carCode STRING,\n" +
                        "    ctype STRING,\n" +
                        "    PRIMARY KEY (id) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        "    'connector' = 'upsert-kafka',\n" +
                        "    'topic' = 'dwd_summary_cars_log',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'key.format' = 'json',\n" +
                        "    'value.format' = 'json'\n" +
                        ")"
        );
    }

    private static Table handle(TableEnvironment stream) {
        // todo 查询
        Table searchLogTable = stream.sqlQuery(
                "SELECT\n" +
                        "  w.id,\n" +
                        "  COALESCE(d.owerId, -1) AS owerId,\n" +
                        "  w.opTime,\n" +
                        "  w.cId,\n" +
                        "  w.carCode,\n" +
                        "  w.ctype\n" +
                        "FROM\n" +
                        "  ods_cars_log_clean AS w\n" +
                        "LEFT JOIN ods_cars_clean AS d\n" +
                        "ON w.carCode = d.carCode"
        );
        stream.createTemporaryView("searchLogTable", searchLogTable);
        return searchLogTable;
    }


    private static void createInputTable(TableEnvironment stream) {

        // 1 门禁刷卡事实表 ods_entrance_guard_log_clean
        stream.executeSql(
                "CREATE TABLE ods_entrance_guard_log_clean(\n" +
                        "    id STRING NOT NULL,\n" +
                        "    openMode STRING,\n" +
                        "    openTime TIMESTAMP(3),\n" +
                        "    ownerId BIGINT,\n" +
                        "    proc_time as PROCTIME(),\n" +
                        "    WATERMARK FOR openTime AS openTime - INTERVAL '5' DAYS\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'ods-entrance-guard-log',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'properties.group.id' = 'ods_entrance_guard_log_clean',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json'\n" +
                        ")"
        );

        // 2 用户维表 ods_ower_clean
        stream.executeSql(
                "CREATE TABLE ods_ower_clean(\n" +
                        "    id BIGINT NOT NULL,\n" +
                        "    roomId BIGINT NOT NULL,\n" +
                        "    owerCode STRING,\n" +
                        "    owerName STRING,\n" +
                        "    owerSex STRING,\n" +
                        "    owerAge BIGINT,\n" +
                        "    owerIdcard STRING,\n" +
                        "    owerTel BIGINT,\n" +
                        "    owerHomeplace STRING,\n" +
                        "    owerWorkplace STRING,\n" +
                        "    remark STRING,\n" +
                        "    createTime TIMESTAMP(3),\n" +
                        "    operatorTime TIMESTAMP(3),\n" +
                        "    proc_time as PROCTIME(),\n" +
                        "    WATERMARK FOR createTime AS createTime - INTERVAL '5' DAYS\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'ods-ower-info',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'properties.group.id' = 'ods_ower_clean',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json'\n" +
                        ")"
        );

        // 3 房间维表 ods_room_clean
        stream.executeSql(
                "CREATE TABLE ods_room_clean(\n" +
                        "    id BIGINT NOT NULL,\n" +
                        "    houseId BIGINT NOT NULL,\n" +
                        "    roomCode STRING,\n" +
                        "    roomName STRING,\n" +
                        "    peopleNum BIGINT,\n" +
                        "    remark STRING,\n" +
                        "    proc_time as PROCTIME()\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'ods-room-info',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'properties.group.id' = 'ods_room_clean',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json'\n" +
                        ")"
        );

        // 4 楼栋维表 house
        stream.executeSql(
                "CREATE TABLE ods_house (\n" +
                        "    id BIGINT,\n" +
                        "    cid BIGINT,\n" +
                        "    houseCode STRING,\n" +
                        "    houseName STRING,\n" +
                        "    buildStartTime STRING,\n" +
                        "    buildEndTime STRING,\n" +
                        "    houseArea STRING,\n" +
                        "    remark STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'sca_doris.house',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );

        // 5 社区维表 community
        stream.executeSql(
                "CREATE TABLE ods_community (\n" +
                        "    id BIGINT,\n" +
                        "    communityName STRING,\n" +
                        "    district_id BIGINT,\n" +
                        "    city_id BIGINT,\n" +
                        "    communityLocation STRING,\n" +
                        "    remark STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'sca_two.community',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );

        // 6 地区维表 district
        stream.executeSql(
                "CREATE TABLE ods_district (\n" +
                        "    id BIGINT,\n" +
                        "    pid BIGINT,\n" +
                        "    code STRING,\n" +
                        "    name STRING,\n" +
                        "    remark STRING,\n" +
                        "    create_time STRING,\n" +
                        "    update_time STRING,\n" +
                        "    status STRING,\n" +
                        "    level STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'sca_two.district',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );
        
    }

    public static TableEnvironment getTableEnv() {
        // 1.环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment stream = TableEnvironment.create(settings);
        // 2.配置属性设置
        Configuration configuration = stream.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        configuration.setString("execution.checkpointing.interval", "5 s");
        // 3.返回对象
        return stream;
    }

}
